package com.xxwu.messagequeue.mqserver.datacenter;

import com.xxwu.messagequeue.common.MqException;
import com.xxwu.messagequeue.mqserver.core.Binding;
import com.xxwu.messagequeue.mqserver.core.Exchange;
import com.xxwu.messagequeue.mqserver.core.MSGQueue;
import com.xxwu.messagequeue.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 实现内存管理数据
 */
@Slf4j
public class MemoryDataCenter {
    // 设计各个组件存储的数据结构
    // 交换机：key = exchangeName, value = exchange 对象
    private ConcurrentHashMap<String, Exchange> exchangesMap = new ConcurrentHashMap<>();
    // 队列:key = queueName, value = queue 对象
    private ConcurrentHashMap<String, MSGQueue> queuesMap = new ConcurrentHashMap<>();
    // 绑定：key1 = exchangeName,value 中的key2 = queueName , value = binding
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 消息：key = messageId, value = message 对象
    private ConcurrentHashMap<String, Message> messagesMap = new ConcurrentHashMap<>();
    // 由于消息是存储在队列中，这里还要建立消息到队列的映射关系
    // 但注意，虽然在 messageMap 中将 message 存储了，又在 messageQueueMap 将 message 存储了一遍
    // 但存储的都是同一个 message 对象，所以并没有占用两份内存空间
    // key = queueName, value = messages
    private ConcurrentHashMap<String, List<Message>> messageQueueMap = new ConcurrentHashMap<>();
    // 这里还有一个未确认消息队列
    // key1 = queueName, key2 = messageId, value = message对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> messageQueueWaitAckMap = new ConcurrentHashMap<>();

    // 添加交换机
    public void insertExchange(Exchange exchange) {
        exchangesMap.put(exchange.getName(), exchange);
        log.info("[MemoryDataCenter] 添加交换机成功， exchangeName={}", exchange.getName());
    }
    // 获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangesMap.get(exchangeName);
    }
    // 删除交换机
    public void deleteExchange(String exchangeName) {
        exchangesMap.remove(exchangeName);
        log.info("[MemoryDataCenter] 删除交换机成功 exchangeName={}", exchangeName);
    }

    // 添加队列
    public void insertQueue(MSGQueue queue) {
        queuesMap.put(queue.getName(), queue);
        log.info("[MemoryDataCenter] 添加队列成功 queueName={}", queue.getName());
    }
    // 获取队列
    public MSGQueue getQueue(String queueName) {
        return queuesMap.get(queueName);
    }
    // 删除队列
    public void deleteQueue(String queueName) {
        queuesMap.remove(queueName);
        log.info("[MemoryDataCenter] 删除队列成功 queueName={}", queueName);
    }

    // 添加绑定
    public void insertBinding(Binding binding) throws MqException {
        // 若通过 exchangeName 查询到的 value 不存在，要先创建才能添加
        // computeIfAbsent 可先验证 value 是否存在，若不存在就先创建一个
        ConcurrentHashMap<String, Binding> bindingMap =
                bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
        // 这里设计到线程安全问题，若线程a、b同时同一个绑定对象,exchangeName 和 queueName 都相同，但bindingKey 不相同
        // 线程a 、b都判断该对象不存在
        // 线程a判断完后，线程b先进行插入了，而对于HashMap来说，一个Key只能对应一个value
        // 所以当线程a进行插入时，就会造成数据覆盖
        // 所以进行加锁
        synchronized (bindingMap) {
            if(bindingMap.containsKey(binding.getQueueName())) {
                throw new MqException("绑定已存在 exchangeName=" + binding.getQueueName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
            log.info("添加交换机成功 exchangeName={}, queueName={}"
                    , binding.getExchangeName(), binding.getQueueName());
        }
    }
    // 获取绑定
    // 这里写两种方法
    // 1 根据 exchangeName 获取所有绑定
    public ConcurrentHashMap<String, Binding> getBindings(String queueName) {
        return bindingsMap.get(queueName);
    }
    // 2 根据 exchangeName 和 queueName 获取唯一的 binding 对象
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = getBindings(exchangeName);
        if(bindingMap == null){
            return null;
        }
        return bindingMap.get(queueName);
    }
    // 删除绑定
    public void deleteBinding(String exchangeName, String queueName) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = getBindings(exchangeName);
        if(bindingMap == null){
            throw new MqException("要删除的交换机对象不存在，exchangeName=" + exchangeName
                    + "queueName=" + queueName);
        }
        bindingMap.remove(queueName);
        log.info("[MemoryDataCenter] 删除交换机成功 exchangeName={}, queueName={}", exchangeName, queueName);
    }

    // 添加消息
    public void insertMessage(Message message) {
        messagesMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 添加消息成功,message={}",message.getMessageId());
    }
    // 获取消息
    public Message getMessage(String messageId) {
        return messagesMap.get(messageId);
    }
    // 删除消息
    public void deleteMessage(String messageId) {
        messagesMap.remove(messageId);
        log.info("[MemoryDataCenter] 删除消息成功,messageId={}", messageId);
    }

    //添加消息到消息队列
    public void sendMessageQueue(String queueName, Message message){
        //若没有 list 则要先创建
        List<Message> messageList = messageQueueMap.computeIfAbsent(queueName
                , k -> new ArrayList<>());
        // 链表是线程不安全的，这里要进行加锁，否则会造成数据覆盖
        synchronized (messageList) {
            messageList.add(message);
        }
        // 将消息在消息工厂中也要添加一份
        messagesMap.put(queueName, message);
        log.info("[MemoryDataCenter] 消息添加到队列中成功 queueName={}, messageId={}"
                , queueName, message.getMessageId());
    }
    // 从消息队列中获取消息
    public Message getMessageQueue(String queueName){
        List<Message> messageList = messageQueueMap.get(queueName);
        // 这里这样写不对，因为 messageList.size() 也可能涉及到线程安全
//        if(messageList == null || messageList.size() == 0){
//            return null;
//        }
        if(messageList == null){
            return null;
        }
        synchronized (messageList) {
            if(messageList.size() == 0){
                return null;
            }
            // 直接将队头元素取出，无需保留
            Message message = messageList.remove(0);
            log.info("[MemoryDataCenter] messageList.size() 获取队头元素 messageId={}", message.getMessageId());
            return message;
        }
    }
    // 获取消息数量
    public int getMessageNums(String queueName){
        List<Message> messageList = messageQueueMap.get(queueName);
        if(messageList == null){
            return 0;
        }
        synchronized (messageList) {
            return messageList.size();
        }
    }

    // 针对为确认消息的处理
    // 添加待确认消息
    public void insertMessageQueueWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> map = messageQueueWaitAckMap.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        map.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 待确认消息添加到指定队列成功 queueName={}, messageId={}", queueName, message.getMessageId());
    }
    // 删除代确认消息(确认后就可删除)
    public void deleteMessageQueueWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> map = messageQueueWaitAckMap.get(queueName);
        map.remove(message.getMessageId());
        log.info("[MemoryDataCenter] 待确认消息删除成功 queueName={}, messageId={}", queueName, message.getMessageId());
    }
    // 获取待确认消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> map = messageQueueWaitAckMap.get(queueName);
        if(map == null){
            return null;
        }
        return map.get(messageId);
    }

    // 恢复所有数据
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {

//        DiskDataCenter diskDataCenter = new DiskDataCenter();

        exchangesMap.clear();
        queuesMap.clear();
        bindingsMap.clear();
        messagesMap.clear();
        messageQueueMap.clear();

        // 恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.getAllExchanges();
        for(Exchange exchange : exchanges) {
            exchangesMap.put(exchange.getName(), exchange);
        }
        // 恢复队列数据
        List<MSGQueue> queues = diskDataCenter.getAllMSGQueues();
        for(MSGQueue queue : queues) {
            queuesMap.put(queue.getName(), queue);
        }
        // 恢复绑定数据
        List<Binding> bindings = diskDataCenter.getAllBindings();
        for(Binding binding : bindings) {
            // 这里不用再把最外层的 HashMap 放进去，在下面这一句就已经根据 exchangeName 创建好了一个键值对放了进去
            // 只需要将内部 HashMap 的键值对放进去即可
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
        }
        // 恢复消息，根据队列名
        for(MSGQueue queue : queues) {
            List<Message> messages = diskDataCenter.getAllMessages(queue.getName());
            // 将消息放入队列消息对应的 HashMap 中
            messageQueueMap.put(queue.getName(), messages);
            // 将消息放入消息大厅的 Map 中
            for(Message message : messages) {
                messagesMap.put(message.getMessageId(), message);
            }
        }
        log.info("文件恢复完成");
        // 针对为确认的消息这里不做处理
        // 在等待服务器等待 ack 的过程中若客户端发生了断线，那么未确认的消息就恢复到消息大厅和消息队列 map 中
        // 需要客户端重新取出进行确认
    }
}
